Developer CD Series 2000 November: Tool Chest/Dev.CD Nov 00 TC Disk 1.toast/Sample Code/Networking/TPIFile/TPIFile.c

Text File | 2000-09-28 | 59.2 KB | 1,830 lines | [TEXT/CWIE]

/*
    File:        TPIFile.c

    Contains:    TPI Module to access File Manager files.  Technology
                demonstration only!

    Written by:    Quinn "The Eskimo!"

    Copyright:    © 1997 by Apple Computer, Inc., all rights reserved.

    Change History (most recent first):

    You may incorporate this sample code into your applications without
    restriction, though the sample code has been provided "AS IS" and the
    responsibility for its operation is 100% yours.  However, what you are
    not permitted to do is to redistribute the source as "DSC Sample Code"
    after having made changes. If you're going to re-distribute the source,
    we require that you make it clear in the source that the code was
    descended from Apple Sample Code, but that you've made changes.

    More assertions than an email meeting with Brian Bechtel!
*/

/////////////////////////////////////////////////////////////////////
// The OT debugging macros in <OTDebug.h> require this variable to
// be set.

#ifndef qDebug
#define qDebug    1
#endif

/////////////////////////////////////////////////////////////////////
// Determine whether this is going to be an instrumented build or not.

#ifndef INSTRUMENTATION_ACTIVE
    #define INSTRUMENTATION_ACTIVE 0
#else
    #define INSTRUMENTATION_ACTIVE 1
#endif

/////////////////////////////////////////////////////////////////////
// Pick up all the standard OT module stuff.

#include <OpenTptModule.h>

/////////////////////////////////////////////////////////////////////
// You've gotta wonder why noCacheMask is only declared in <FSM.h>!

#include <FSM.h>
#include <Files.h>
#include <Devices.h>

/////////////////////////////////////////////////////////////////////
// Pick up Instrumentation SDK stuff.  We only do this if we're
// actually instrumenting, so you don't even have to have the SDK
// to compile the non-instumented version of the code.  If we're
// instrumenting, we compile a bunch of bogus macros that generally
// compile to nothing.

#if INSTRUMENTATION_ACTIVE
    #include <InstrumentationMacros.h>
#else
    #define TRACE_SETUP        long __junk
    #define LOG_ENTRY(n)    if (0) { __junk ; }
    #define LOG_EXIT        if (0) { __junk ; }
#endif

/////////////////////////////////////////////////////////////////////
// Pick up our module specific data structures, specifically
// the AF_FILESPEC address format.

#include "TPIFile.h"

/////////////////////////////////////////////////////////////////////

extern pascal void OTDebugStr(const char *str)
    // OTDebugStr seems to be missing from the latest OpenTptMiscUtilsPPC.o,
    // so we implement our own dummy version here.
{
    debugstr(str);
}

/////////////////////////////////////////////////////////////////////

#if INSTRUMENTATION_ACTIVE

    // If we're instrumenting, do a special hack to allow us
    // to see calls to the File Manager.  We declare stub
    // routines with a trailing "X" for each File Manager
    // routine we call, and then macro define the corresponding
    // identifiers to expand out to the "X" routine.
    //
    // In general you don't have to do this sort of rubbish
    // when instrumenting your code.  The Instrumentation SDK
    // describes how you can use MrProf to do an equivalent
    // thing automagically.  I tried that and TPIFile started
    // crashing unexpectedly.  I would love to know why, but
    // alas I didn't have enough time to debug it.  Instead,
    // I did this hack.  *sigh*

    static pascal OSErr PBHOpenAsyncX(HParmBlkPtr paramBlock)
    {
        TRACE_SETUP;
        OSErr result;

        LOG_ENTRY( "TPIFile:PBHOpenAsync" );
        result = PBHOpenAsync(paramBlock);
        LOG_EXIT;
        return (result);
    }

    static pascal OSErr PBReadAsyncX(ParmBlkPtr paramBlock)
    {
        TRACE_SETUP;
        OSErr result;

        LOG_ENTRY( "TPIFile:PBReadAsync" );
        result = PBReadAsync(paramBlock);
        LOG_EXIT;
        return (result);
    }

    static pascal OSErr PBCloseAsyncX(ParmBlkPtr paramBlock)
    {
        TRACE_SETUP;
        OSErr result;

        LOG_ENTRY( "TPIFile:PBCloseAsync" );
        result = PBCloseAsync(paramBlock);
        LOG_EXIT;
        return (result);
    }

    #define PBHOpenAsync    PBHOpenAsyncX
    #define PBReadAsync        PBReadAsyncX
    #define PBCloseAsync    PBCloseAsyncX

#endif

/////////////////////////////////////////////////////////////////////

static mblk_t* qmi_tpi_data_ind(mblk_t* trailer_mp, int flags, long type)
    // I found that mi_tpi_data_ind is not exported by the OT libraries,
    // so I simply include my own verison here.
{
    mblk_t* mp;

    mp = mi_tpi_data_req(trailer_mp, flags, type);
    if (mp)
        ((struct T_data_ind *)mp->b_rptr)->PRIM_type = T_DATA_IND;
    return mp;
}

/////////////////////////////////////////////////////////////////////

static Boolean IsReadQ(queue_t* q)
    // Returns true if q is the read queue of a queue pair.
{
    return ( (q->q_flag & QREADR) != 0 );
}

static Boolean IsWriteQ(queue_t* q)
    // Returns true if q is the write queue of a queue pair.
{
    return ( (q->q_flag & QREADR) == 0 );
}

/////////////////////////////////////////////////////////////////////

// kDataBufferSize is the number of bytes we read of the file
// at a time.  If you have a file that's larger than kDataBufferSize,
// it gets sent upstream in kDataBufferSize chunks.  The number was
// chosen because I know that OT doesn't like you asking for big
// large individual memory buffers.  Remember, OT's memory manager
// is designed for networking, and optimised for buffers that are
// the size of your typical network packet.

enum {
    kDataBufferSize = 2048
};

/////////////////////////////////////////////////////////////////////
// Per-Stream information

// This structure is used to hold the per-stream data for the module.
// While module's can use normal global variables to store real globals,
// they must maintain their own per-stream data structures.  I use
// mi_open_comm to allocate this data structure when the stream is
// opened.  mi_open_comm stores the address of this data structure in the
// read and write queue's q_ptr field, so the rest of the code
// can get to it by calling the GetPerStreamData function.

    // Due to an intensely annoying inability to distinguish three different
    // states in DoDisconnectRequestAck, I need to define a sub-state variable
    // that distinguishes between the possible ways to get to DoDisconnectRequestAck
    // when the stream is in state TS_WACK_DREQ6.

    typedef enum {
        WSS_OPEN = 0,            // Got to DoDisconnectRequestAck by getting a T_DISCON_REQ while in TS_WCON_CREQ.
        WSS_ISSUE_MP            // Got to DoDisconnectRequestAck by calling CloseFileAndPostMessage.
    } WackSubStateType;

struct PerStreamData
{
    OSType                 magic;                // kTPIFilePerStreamDataMagic = 'ESK0' for debugging

    long                   currentState;        // State of the TPI module, TS_UNBND etc

    WackSubStateType    wackSubstate;        // Sub-state while in TS_WACK_DREQ6, WSS_OPEN etc

    queue_t             *readQueue;            // Read queue for this stream.  Used by the
                                            // the ioCompletion routine to find the queue.

    mblk_t                 *currentMessage;    // Current message pending ioCompletion.
                                            // When the CustomIOCompletion routine fires,
                                            // it puts this message on the above queue.

    HParamBlockRec        fileParamBlock;        // Current pending IOParamBlock

    FSSpec                 fileSpec;            // The file we're connected to.

    short                 fileRefNum;            // The fileRefNum if the file is open.

    Boolean                detached;            // Whether this stream has been detached.
                                            // Streams get detached when they are closed
                                            // but there is outstanding asynchronous I/O
                                            // which we have to wait to complete before
                                            // we can dispose of this data structure.
};
typedef struct PerStreamData PerStreamData, *PerStreamDataPtr;


static PerStreamDataPtr GetPerStreamData(queue_t* readOrWriteQ)
    // You can pass both the read or the write queue to this routine
    // because mi_open_comm sets up both q_ptr's to point to the
    // queue local data.
    //
    // Note that, in order to avoid the overhead of a function call,
    // you would normally use inline code (or a macro)
    // to get your per-stream data instead of using a separate function.
    // However I think the separate function makes things clearer.
    // I also acts as a central bottleneck for my debugging code.
    //
    // Environment: any standard STREAMS entry point
{
    PerStreamDataPtr streamData;

    streamData = (PerStreamDataPtr) readOrWriteQ->q_ptr;

    OTAssert("GetPerStreamData: what streamData", streamData != nil);
    OTAssert("GetPerStreamData: Bad magic", streamData->magic == kTPIFilePerStreamDataMagic);

    return (streamData);
}

// mi_open_comm and mi_close_comm (and also mi_detach and mi_close_detached)
// use this global to store the list of open streams to this module.

static char* gModuleList = nil;

// A UPP for CustomIOCompletion.

static IOCompletionUPP gCustomIOCompletionUPP = nil;

/////////////////////////////////////////////////////////////////////

static void CloseFileAndPostMessage(queue_t* q, mblk_t* mp)
    // This routine is called to close the open file associated
    // with the stream the given queue (which is the read-side
    // queue).  mp is either nil, or the message that should be
    // posted to the read-side service routine when the close is
    // complete.
    //
    // Environment: read service routine
    // Environment: Deferred Task (called by CleanUpPerStreamData)
    // Environment: close entry point
{
    TRACE_SETUP;
    PerStreamDataPtr streamData;

    LOG_ENTRY( "TPIFile:CloseFileAndPostMessage" );
    OTAssert("CloseFileAndPostMessage: Not the read queue", IsReadQ(q) );

    streamData = GetPerStreamData(q);
    OTAssert("CloseFileAndPostMessage: Already have a current message", streamData->currentMessage == nil);
    OTAssert("CloseFileAndPostMessage: File isn't open", streamData->fileRefNum != 0);

    streamData->currentMessage = mp;

    // Set up the parameter block for the _Close.

    streamData->fileParamBlock.ioParam.ioRefNum = streamData->fileRefNum;
    streamData->fileParamBlock.ioParam.ioCompletion = gCustomIOCompletionUPP;

    // This is critical.  Set our copy of file reference number to 0 so
    // that DoDisconnectRequestAck knows that it's the second time it's been called.

    streamData->fileRefNum = 0;

    // Start the close operation.

    (void) PBCloseAsync( (ParmBlkPtr) &streamData->fileParamBlock);

    // ... continue in CustomIOCompletion... TPIFileReadService... ( DoDisconnectRequestAck or DoDisconnectIndication)...
    LOG_EXIT;
}

/////////////////////////////////////////////////////////////////////

static Boolean CleanUpPerStreamData(PerStreamDataPtr streamData)
    // This routine is called to clean up the per-stream data for this
    // stream.  The routine returns true if the stream is clean (ie
    // the memory can be disposed of), or false if there is an
    // async close operation that means the stream must be left lying around.
    // If it returns false, the rest of the module organises to call
    // it repeatably until it returns true.
    //
    // Environment: Deferred Task (scheduled by CustomIOCompletion)
    // Environment: close entry point
    // Environment: TerminateStreamModule
{
    TRACE_SETUP;
    Boolean result;

    LOG_ENTRY( "TPIFile:CleanUpPerStreamData" );

    OTAssert("CleanUpPerStreamData: Already have a current message", streamData->currentMessage == nil);

    // Test to see whether the file is currently open.

    if ( (streamData->fileRefNum != 0) || (streamData->fileParamBlock.ioParam.ioResult > noErr) ) {

        // The file is open, or the close is still in progress.  If the close isn't
        // in progress, start one.  Regardless, return false to let the caller know
        // we can't be killed yet.

        if ( streamData->fileParamBlock.ioParam.ioResult <= noErr ) {
            // OTDebugBreak("CleanUpPerStreamData: Executing the especially hard case");
            CloseFileAndPostMessage(streamData->readQueue, nil);
        }

        result = false;
    } else {

        // The file is closed.  No other clean up required, but if streamData
        // contained pointers to other data structures, this is where we would
        // dispose them.

        result = true;
    }

    LOG_EXIT;
    return (result);
}

/////////////////////////////////////////////////////////////////////

static long gCleanUpAnyDetachedStreamsTaskID = 0;

static pascal void CleanUpAnyDetachedStreams(void* junkArg)
    // Loop through all of the streams looking for detached ones.
    // When we find a detached stream, attempt to close it by
    // calling CleanUpPerStreamData.
    //
    // When the ioCompletion fires but finds it hasn't got a
    // message to work with (because the stream has closed) it
    // schedules this deferred task to run and actually kill
    // the stream.
    //
    // TerminateStreamModule also calls this routine
    // to kill any remaining streams before the module quits.
    //
    // Environment: Deferred Task (scheduled by CustomIOCompletion)
    // Environment: TerminateStreamModule
{
    TRACE_SETUP;
    #pragma unused(junkArg)
    PerStreamDataPtr thisStreamData;

    LOG_ENTRY( "TPIFile:CleanUpAnyDetachedStreams" );

    // OTDebugBreak("CleanUpAnyDetachedStreams: Entering");

    // Loop through each of the streams...

    thisStreamData = (PerStreamDataPtr) gModuleList;
    while (thisStreamData != nil) {
        if (thisStreamData->detached && CleanUpPerStreamData( thisStreamData ) ) {

            // Yes're allowed to kill this detached stream.

            mi_close_detached(&gModuleList, (char *) thisStreamData);

            // Resume the search at the beginning of the stream list because
            // a) I don't know the exact ordering that mi_open_comm uses to
            //    create the stream list, so I'll just do the entire thing again.
            // b) There's a possibility that hardware interrupts are detaching
            //    streams while we're working here, so we might as well pick
            //    them up in this pass.

            thisStreamData = (PerStreamDataPtr) gModuleList;

        } else {

            // This stream is not detached or not eligible for closing, continue
            // on with the next one.
            thisStreamData = (PerStreamDataPtr) mi_next_ptr( (char *) thisStreamData );
        }

    }

    // You might think that we can assert that we've mi_close_detached
    // at least one stream here, but that may not be true.  Imagine this
    // scenario:
    //
    // 1. CustomIOCompletion fires, marks stream detached, schedules
    //    CleanUpAnyDetachedStreams
    // 2. CleanUpAnyDetachedStreams runs, finds marked stream,
    //    and calls mi_close_detached on it.  It then resumes its
    //    search at the beginning of the stream list.
    // 3. CustomIOCompletion fires again, for a different stream.
    //    It marks the stream detached and schedules
    //    CleanUpAnyDetachedStreams again.  This works because
    //    OTScheduleDeferredTask considers a running task (ie
    //    CleanUpAnyDetachedStreams) to not be scheduled, so it
    //    will schedule it again.
    // 4. The first instance of CleanUpAnyDetachedStreams continues
    //    running and picks up the second marked stream, which it closes.
    // 5. The second instance of CleanUpAnyDetachedStreams runs,
    //    and can't find a stream to close.
    //
    // The fact that OTScheduleDeferredTask will reschedule a running
    // task is *good* because, although it means there's a possibility
    // that we'll run CleanUpAnyDetachedStreams unnecessarily, it also
    // guarantees that we'll never leave a dangling detached stream.

    LOG_EXIT;
}

/////////////////////////////////////////////////////////////////////

static pascal void CustomIOCompletion(ParmBlkPtr paramBlock)
    // This function is the ioCompletion routine we use for all of
    // our File Manager operations.  Because it operates at hardware
    // interrupt level, it is not synchronised with the rest of STREAMS
    // and so we have to be careful that we only call STREAMS routines
    // that are accessible from hardware interrupt level.  This list
    // of routines is given in the "Open Tpt Module Dev. Note".
    //
    // The routine performs two different functions based on whether
    // currentMessage is nil or not.  If currentMessage is not nil,
    // the routine simply puts that message on the stream's read-side
    // queue.  This schedules the TPIFileReadService, which continues
    // the processing of the operation based on the type of the message.
    // The person who called the File Manager will have set currentMessage
    // up appropriately.
    //
    // If currentMessage is nil, the stream was closed before the
    // File Manager ioCompletion routine ran.  We have to take special
    // care to clean up this stream correctly by scheduling the
    // CleanUpAnyDetachedStreams deferred task.
    //
    // In looking at trace logs it seems that PBCloseAsync pretty much
    // always operates synchronously, so this second case is most probably
    // not well tested.
    //
    // Environment: Hardware Interrupt (ioCompletion from File Manager)
{
    TRACE_SETUP;
    PerStreamDataPtr streamData;
    mblk_t *mp;

    LOG_ENTRY( "TPIFile:CustomIOCompletion" );

    // Tell OT we're operating at hardware interrupt level.

    OTEnterInterrupt();

    // Get the per-stream data associated with this File Manager
    // operation using the standard "offset from the ParamBlockRec"
    // technique.

    streamData = (PerStreamDataPtr) ( ((UInt8 *) paramBlock) - OTOffsetOf(PerStreamData, fileParamBlock) );
    OTAssert("CustomIOCompletion: Bad magic", streamData->magic == kTPIFilePerStreamDataMagic);
    // OTAssert("CustomIOCompletion: This is not wrong, merely interesting", streamData->currentMessage != nil);

    mp = streamData->currentMessage;

    if (mp == nil) {

        // The stream was closed before the ioCompletion routine could fire.
        // Schedule our deferred task to clean up the wreckage.

        OTAssert("CustomIOCompletion: Clean up deferred task not created", gCleanUpAnyDetachedStreamsTaskID != 0);

        streamData->detached = true;

        OTScheduleDeferredTask(gCleanUpAnyDetachedStreamsTaskID);

        // ... continue in CleanUpAnyDetachedStreams...

    } else {

        streamData->currentMessage = nil;

        // The stream is still a going proposition.  Put
        // the current message on our read-side queue,
        // which has the effect of scheduling our read-side
        // service routine TPIFileReadService which will
        // forward the message upstream.

        putq(streamData->readQueue, mp);

        // ... continue in TPIFileReadService ...
    }

    OTLeaveInterrupt();

    LOG_EXIT;
}

/////////////////////////////////////////////////////////////////////
// Open routine

static SInt32 TPIFileOpen(queue_t* rdq, dev_t* dev, SInt32 flag, SInt32 sflag, cred_t* creds)
    // This routine is called by STREAMS when a new stream is connected to
    // our module.  The bulk of the work here is done by the Mentat helper
    // routine mi_open_comm.
    //
    // Environment: standard STREAMS entry point
{
    TRACE_SETUP;
    SInt32 err;
    PerStreamDataPtr streamData;

    LOG_ENTRY( "TPIFile:TPIFileOpen" );

    OTAssert("TPIFileOpen: Not the read queue", IsReadQ(rdq) );

    err = noErr;

    // If we already have per-stream data for this stream, the stream is being reopened.
    // In that case, we can just return.
    // Note that we can't call GetPerStreamData because it checks that streamData is not nil.

    if ( rdq->q_ptr != nil ) {
        goto done;
    }

    // Make sure we're being opened properly -- because we're a driver we
    // don't allow a "module" open.

    if ( err == noErr && sflag == MODOPEN ) {
        err = ENXIO;
    }

    // Use the mi_open_comm routine to allocate our per-stream data.  Then
    // zero out the entire per-stream data record and fill out the fields
    // we're going to need.

    if (err == noErr) {
        err = mi_open_comm(&gModuleList, sizeof(PerStreamData), rdq, dev, flag, sflag, creds);
        if ( err == noErr ) {
            // Note that we can't call GetPerStreamData because the magic is not set up yet.
            streamData = (PerStreamDataPtr) rdq->q_ptr;

            OTMemzero(streamData, sizeof(PerStreamData));

            streamData->magic = kTPIFilePerStreamDataMagic;
            streamData->currentState = TS_UNBND;
            streamData->readQueue = rdq;
        }
    }

done:
    LOG_EXIT;
    return (err);
}

/////////////////////////////////////////////////////////////////////
// Close routine

static SInt32 TPIFileClose(queue_t* rdq, SInt32 flags, cred_t* credP)
    // This routine is called by STREAMS when a stream is being
    // disconnected from our driver (ie closed).  The operation of this
    // routine is complicated by one important fact: because we
    // call File Manager asynchronously, there may be a File Manager
    // I/O operation in progress when the stream is closed.  If this
    // is the case, must use a special technique (mi_detach) to
    // detach our per-stream data from the actual stream, and
    // then organise to clean up this data at some later stage.
    //
    // Environment: standard STREAMS entry point
{
    TRACE_SETUP;
    #pragma unused(flags)
    #pragma unused(credP)
    PerStreamDataPtr streamData;
    mblk_t *oldCurrentMessage;

    LOG_ENTRY( "TPIFile:TPIFileClose" );
    OTAssert("TPIFileClose: Not the read queue", IsReadQ(rdq) );

    streamData = GetPerStreamData(rdq);

    // Be very careful here.  streamData->currentMessage is also modified by
    // the CustomIOCompletion routine, which (unlike the rest of our
    // module) is not synchronised with this routine.  So we must use
    // atomic operations to do this change streamData->currentMessage
    // to make sure that either we close the stream or we detach the
    // stream and CustomIOCompletion uses a deferred task to close it.
    //
    // Note that CustomIOCompletion does not have to use an atomic
    // operation to modify streamData->currentMessage because it can
    // interrupt us but we can't interrupt it.

    do {
        oldCurrentMessage = streamData->currentMessage;
    } while ( ! OTCompareAndSwapPtr(oldCurrentMessage, nil, &streamData->currentMessage) );

    if ( oldCurrentMessage != nil ) {

        // The ioCompletion routine hasn't fired yet, and if it does
        // we've atomically guaranteed that it will know that this
        // queue has been detached.  So we detach the queue and
        // wait for the completion routine to schedule our deferred
        // task.

        // OTDebugBreak("TPIFileClose: Executing the hard case");

        freemsg(oldCurrentMessage);

        mi_detach(rdq, (char *) streamData);

        // ... continue in CustomIOCompletion...

    } else {

        if ( CleanUpPerStreamData( streamData ) ) {

            // The easy case.  There is no outstanding ioCompletion routine,
            // and the file is closed, so we can just shut down this stream.

            (void) mi_close_comm(&gModuleList, rdq);

        } else {

            // CleanUpPerStreamData has scheduled an async _Close,
            // detach the stream and wait for it.

            mi_detach(rdq, (char *) streamData);

            // ... continue in CustomIOCompletion...
        }

    }

    LOG_EXIT;
    return (0);
}

/////////////////////////////////////////////////////////////////////

enum {
    kNoPrimitive = -1
};

static long GetPrimitive(mblk_t* mp)
    // GetPrimitive gets the TPI primitive out of a message block.
    // It returns kNoPrimitive if the message block is of the wrong
    // type or there is no TPI primitive.
    //
    // Environment: any standard STREAMS entry point
{
    if ((mp->b_datap->db_type == M_PROTO || mp->b_datap->db_type == M_PCPROTO) && MBLK_SIZE(mp) >= sizeof(long) ) {
        return ( ( (union T_primitives*) mp->b_rptr)->type );
    } else {
        return ( kNoPrimitive );
    }
}

/////////////////////////////////////////////////////////////////////

static void DoInfoRequest(queue_t* q, mblk_t* mp)
    // Handle a T_INFO_REQ TPI message by responding with
    // a T_INFO_ACK message.
    //
    // Environment: write put routine
{
    TRACE_SETUP;
    T_info_ack *infoAck;
    PerStreamDataPtr streamData;

    LOG_ENTRY( "TPIFile:DoInfoRequest" );
    OTAssert("DoInfoRequest: Not the write queue", IsWriteQ(q) );

    streamData = GetPerStreamData(q);

    // Allocate the T_INFO_ACK message, reusing mp if possible.

    mp = mi_tpi_ack_alloc(mp, sizeof(T_info_ack), T_INFO_ACK);
    OTAssert("mi_tpi_ack_alloc failed", mp != nil );

    infoAck = (T_info_ack *) mp->b_rptr;

    // Fill out infoAck.  Note that we do not say we support
    // orderly release!

    infoAck->TSDU_size = 0;
    infoAck->ETSDU_size = T_INVALID;
    infoAck->CDATA_size = T_INVALID;
    infoAck->DDATA_size = T_INVALID;
    infoAck->ADDR_size = T_INFINITE;
    infoAck->OPT_size = T_INVALID;
    infoAck->TIDU_size = T_INFINITE;
    infoAck->SERV_type = T_COTS;
    infoAck->CURRENT_state = streamData->currentState;
    infoAck->PROVIDER_flag = 0;

    (void) qreply(q, mp);
    LOG_EXIT;
}

/////////////////////////////////////////////////////////////////////

static void ReplyWithErrorAck(queue_t* q, mblk_t* mp, long tpiError, long unixError)
    // A simple routine which replies to a TPI message with a T_ERROR_ACK
    // message containing the error codes tpiError and unixError.
    //
    // Environment: write put routine
{
    TRACE_SETUP;

    LOG_ENTRY( "TPIFile:ReplyWithErrorAck" );
    OTAssert("ReplyWithErrorAck: Not the write queue", IsWriteQ(q) );

    mp = mi_tpi_err_ack_alloc(mp, tpiError, unixError);
    OTAssert("mi_tpi_err_ack_alloc failed", mp != nil );
    (void) qreply(q, mp);
    LOG_EXIT;
}

/////////////////////////////////////////////////////////////////////

static void DoBindRequest(queue_t* q, mblk_t* mp)
    // Handle a T_BIND_REQ message by changing to the bound state
    // and replying with a T_BIND_ACK message.
    //
    // Environment: write put routine
{
    TRACE_SETUP;
    T_bind_req *bindReq;
    T_bind_ack *bindAck;
    PerStreamDataPtr streamData;

    LOG_ENTRY( "TPIFile:DoBindRequest" );
    OTAssert("DoBindRequest: Not the write queue", IsWriteQ(q) );

    streamData = GetPerStreamData(q);

    // Check whether we're in the right state.
    if (streamData->currentState != TS_UNBND) {
        ReplyWithErrorAck(q, mp, TOUTSTATE, 0);
        goto done;
    }

    // Check the bind parameters.
    bindReq = (T_bind_req *) mp->b_rptr;

    if (bindReq->CONIND_number != 0 || bindReq->ADDR_length != 0) {
        ReplyWithErrorAck(q, mp, TNOADDR, 0);
        goto done;
    }

    // All is cool, lets say we're bound.
    mp = mi_tpi_ack_alloc(mp, sizeof(T_bind_ack), T_BIND_ACK);
    OTAssert("mi_tpi_ack_alloc failed", mp != nil );

    bindAck = (T_bind_ack *) mp->b_rptr;

    bindAck->ADDR_length = 0;
    bindAck->ADDR_offset = sizeof(T_bind_ack);
    bindAck->CONIND_number = 0;

    // Switch to the bound state.

    streamData->currentState = TS_IDLE;

    qreply(q, mp);

done:
    LOG_EXIT;
    return;
}

/////////////////////////////////////////////////////////////////////

static void DoConnectRequest(queue_t* q, mblk_t* mp)
    // Handle a T_CONN_REQ message in two stages.  If the connect
    // request contains bogus information, we immediately NAK
    // it with a T_ERROR_ACK.  If the information in the connect
    // request looks OK, we start the connection process (ie by
    // calling PBHOpenAsync) and send up a T_OK_ACK to say that
    // the connection is in progress.  When the async operation
    // completes, the ioCompletion fires, calling CustomIOCompletion
    // which in turn puts streamData->currentMessage on the read-side
    // queue, which schedule TPIFileReadService, which calls
    // DoConnectConfirm which finally creates the T_CONN_CON message.
    //
    // Environment: write put routine
{
    TRACE_SETUP;
    T_conn_req *connReq;
    FileSpecAddressPtr connAddr;
    PerStreamDataPtr streamData;

    LOG_ENTRY( "TPIFile:DoConnectRequest" );
    OTAssert("DoConnectRequest: Not the write queue", IsWriteQ(q) );

    streamData = GetPerStreamData(q);

    // Check whether we're in the right state.
    if (streamData->currentState != TS_IDLE) {
        ReplyWithErrorAck(q, mp, TOUTSTATE, 0);
        goto done;
    }

    // Check the connect parameters.
    connReq = (T_conn_req *) mp->b_rptr;

    if (connReq->OPT_length != 0) {
        ReplyWithErrorAck(q, mp, TBADOPT, 0);
        goto done;
    }
    if (connReq->DEST_length != sizeof(FileSpecAddress)) {
        ReplyWithErrorAck(q, mp, TBADADDR, 0);
        goto done;
    }
    connAddr = (FileSpecAddressPtr) mi_offset_paramc(mp, connReq->DEST_offset, connReq->DEST_length);
    if (connAddr->fAddressType != AF_FILESPEC) {
        ReplyWithErrorAck(q, mp, TBADADDR, 0);
        goto done;
    }

    streamData->fileSpec = connAddr->fss;

    // Create the T_CONN_CON message for use by the ioCompletion routine.

    OTAssert("DoConnectRequest: Already have a current message", streamData->currentMessage == nil);
    streamData->currentMessage = mi_tpi_conn_con(nil, (char *) connAddr, sizeof(FileSpecAddress), nil, 0);
    OTAssert("DoConnectRequest: mi_tpi_conn_con failed", streamData->currentMessage != nil);

    // mi_tpi_conn_con fills in all the fields of the message, so we have nothing more to do.

    streamData->currentState = TS_WCON_CREQ;

    streamData->fileParamBlock.ioParam.ioVRefNum = streamData->fileSpec.vRefNum;
    streamData->fileParamBlock.fileParam.ioDirID = streamData->fileSpec.parID;
    streamData->fileParamBlock.ioParam.ioNamePtr = &streamData->fileSpec.name[0];
    streamData->fileParamBlock.ioParam.ioPermssn = fsRdPerm;
    streamData->fileParamBlock.ioParam.ioVersNum = 0;
    streamData->fileParamBlock.ioParam.ioMisc = nil;
    streamData->fileParamBlock.ioParam.ioCompletion = gCustomIOCompletionUPP;
    (void) PBHOpenAsync(&streamData->fileParamBlock);
    // Throw away error result from PBOpenAsync because File Manager will still call
    //  our completion routine.

    // We've started the connection attempt, respond with a T_OK_ACK.

    mp = mi_tpi_ok_ack_alloc(mp);
    OTAssert("mi_tpi_ok_ack_alloc failed", mp != nil );

    qreply(q, mp);

    // ... continue in CustomIOCompletion... TPIFileReadService... DoConnectConfirm...

done:
    LOG_EXIT;
    return;
}

/////////////////////////////////////////////////////////////////////

static void DoDisconnectRequest(queue_t* q, mblk_t* mp)
    // Handling a T_DISCON_REQ message is complicated by two
    // factors: a) there are two possible states that the stream
    // can be in and get a valid T_DISCON_REQ, and we have to
    // handle both cases slightly differently, and b) there
    // could be a pending asynchronous I/O operation for this
    // stream, and we don't want it going upstream after we've
    // disconnected.
    //
    // To handle the first difficulty, we have to case our
    // behaviour on the current state of the stream.
    // If the stream is in TS_WCON_CREQ (Waiting for Confirmation
    // of Connection Request -- we were asked to disconnect
    // while we were in the process of connecting), we switch
    // to state TS_WACK_DREQ6.  If the state is TS_DATA_XFER
    // (Data Transfer -- we were asked to disconnect while
    // we are tranferring data), we switch to TS_WACK_DREQ9.
    //
    // This resulting state is important because DoDisconnectRequestAck
    // uses it to determine the next state in the state machine.
    //
    // We can't just send the T_OK_ACK upstream from here because
    // we most probably have an outstanding File Manager completion
    // routine (and it's hard to tell because File Manager completions
    // are not synchronised with STREAMS) for the _Read or _Open,
    // and we don't want that to complete at an unexpected
    // time.  In addition, there is no way of cancelling
    // outstanding File Manager requests.
    //
    // In the case where there is a pending File Manager operation,
    // we simply swap out the current message and replace it with
    // our T_OK_ACK message.  This must be done atomically to ensure
    // that the completion routine doesn't fire while we're doing it.
    //
    // In the case where there is no pending File Manager operation,
    // we post the message to the read-side queue directly.  This
    // especially important in the case when the stream in in
    // TS_DATA_XFER, because the read-side queue may be descheduled
    // because of flow control.  Fortunately the T_OK_ACK message
    // is a high-priority message, so it will be processed anyway.
    //
    // Environment: write put routine
{
    TRACE_SETUP;
    PerStreamDataPtr streamData;
    mblk_t* oldCurrentMessage;

    LOG_ENTRY( "TPIFile:DoDisconnectRequest" );
    // OTDebugBreak("DoDisconnectRequest: Disconnect requested");

    OTAssert("DoDisconnectRequest: Not the write queue", IsWriteQ(q) );

    streamData = GetPerStreamData(q);

    // Do the right thing depending on the current state.

    switch ( streamData->currentState ) {
        case TS_WCON_CREQ:
        case TS_DATA_XFER:

            // Set the new state appropriately, so that
            // DoDisconnectRequestAck knows what to do.

            if ( streamData->currentState == TS_WCON_CREQ ) {
                streamData->wackSubstate = WSS_OPEN;
                streamData->currentState = TS_WACK_DREQ6;
            } else {
                streamData->currentState = TS_WACK_DREQ9;
            }

            // Allocate an appropriate T_OK_ACK message.

            mp = mi_tpi_ok_ack_alloc(mp);
            OTAssert("DoDisconnectRequest: mi_tpi_ok_ack_alloc failed", mp != nil );

            // Swap it into this stream's currentMessage field.

            oldCurrentMessage = streamData->currentMessage;
            if ( (oldCurrentMessage != nil ) &&
                        OTCompareAndSwapPtr(oldCurrentMessage, mp, &streamData->currentMessage) ) {
                // We successfully managed to swap the message into the
                // current message field.  The T_OK_ACK message will be
                // posted when the ioCompletion runs.

                freemsg(oldCurrentMessage);
            } else {
                OTAssert("DoDisconnectRequest: Message unexpectedly appeared", streamData->currentMessage == nil);

                // There is either no outstanding I/O operation, or it fired
                // while we were looking at it.  We must do clever things!
                // namely putting our T_OK_ACK message on the read-side queue.
                //
                // Because this is a high-priority message it will be delivered
                // before the T_CONN_CON or T_DATA_IND messages that might have
                // been queued by the ioCompletion routine before it in the stream
                // read-side queue.

                putq( RD(q) , mp);

                // btw We do not freemsg(oldCurrentMessage) because the interrupt
                // handler has already put it on our read-side queue.
            }

            // ... continue in DoDisconnectRequestAck...

            break;

        default:
            OTDebugBreak("DoDisconnectRequest: Out of state");
            ReplyWithErrorAck(q, mp, TOUTSTATE, 0);
            break;
    }
    LOG_EXIT;
}

/////////////////////////////////////////////////////////////////////

static void DoUnbindRequest(queue_t* q, mblk_t* mp)
    // Handle a T_UNBIND_REQ message.  TPI says that
    // we can only get these in the TS_IDLE idle state.
    // If that's the case, we simply change state to
    // TS_UNBND and reply with a T_OK_ACK.
    //
    // Environment: write put routine
{
    TRACE_SETUP;
    PerStreamDataPtr streamData;

    LOG_ENTRY( "TPIFile:DoUnbindRequest" );
    OTAssert("DoUnbindRequest: Not the write queue", IsWriteQ(q) );

    streamData = GetPerStreamData(q);

    // Check whether we're in the right state.
    if (streamData->currentState != TS_IDLE) {
        ReplyWithErrorAck(q, mp, TOUTSTATE, 0);
        goto done;
    }

    // No other parameters to check, so everything is cool.
    // We change state and ACK the request.

    streamData->currentState = TS_UNBND;

    mp = mi_tpi_ok_ack_alloc(mp);
    OTAssert("DoUnbindRequest: mi_tpi_ok_ack_alloc failed", mp != nil );

    qreply(q, mp);

done:
    LOG_EXIT;
}

/////////////////////////////////////////////////////////////////////

static void DoOptionManagementRequest(queue_t* q, mblk_t* mp)
    // We've told the world that we don't support option management
    // so just error any T_OPTMGMT_REQ messages.
    //
    // Environment: write put routine
{
    TRACE_SETUP;
    OTAssert("DoOptionManagementRequest: Not the write queue", IsWriteQ(q) );

    LOG_ENTRY( "TPIFile:DoOptionManagementRequest" );
    ReplyWithErrorAck(q, mp, TBADOPT, 0);
    LOG_EXIT;
}

/////////////////////////////////////////////////////////////////////

static void DoStreamFatalError(queue_t* q, mblk_t* mp)
    // Send up a stream-fatal M_ERROR message.  We do this in response
    // to messages that are just wrong for our type of stream.
    //
    // Environment: write put routine
{
    TRACE_SETUP;

    OTDebugBreak("DoStreamFatalError");

    LOG_ENTRY( "TPIFile:DoStreamFatalError" );
    OTAssert("DoStreamFatalError: Not the write queue", IsWriteQ(q) );
    OTAssert("DoStreamFatalError: mp is nil", mp != nil );
    freemsg(mp);

    // Allocate a 1 byte M_ERROR message containing EPROTO
    // and send it back up the stream.  This basically kills
    // the stream as far as the client is concerned.  Normally
    // you wouldn't do this except in dire circumstances.  I
    // do it when I find that someone has sent me the wrong
    // kind of message.  Normally the stream head should protect
    // you against this.

    mp = allocb(sizeof(UInt8), 0);
    OTAssert("DoStreamFatalError: allocb failed", mp != nil );
    OTAssert("DoStreamFatalError: allocb is not doing what we expect", mp->b_rptr == mp->b_wptr);

    mp->b_datap->db_type = M_ERROR;

    *((UInt8 *) (mp->b_wptr)) = EPROTO;
    mp->b_wptr = mp->b_rptr + 1;

    qreply(q, mp);
    LOG_EXIT;
}

/////////////////////////////////////////////////////////////////////

static void DoDataRequest(queue_t* q, mblk_t* headerBlock, mblk_t* dataBlock)
    // Handle a T_DATA_REQ message.  Well actually, we don't handle
    // them, so we kill the stream if we get one.
    //
    // headerBlock may be nil when this messages is caused by an M_DATA
    // message.
    //
    // Environment: write put routine
{
    TRACE_SETUP;
    mblk_t* mp;

    LOG_ENTRY( "TPIFile:DoDataRequest" );

    OTAssert("DoDataRequest: Not the write queue", IsWriteQ(q) );

    if ( headerBlock == nil ) {
        mp = dataBlock;
    } else {
        mp = headerBlock;
    }

    // We do not handle writing data to TPIFile streams.
    DoStreamFatalError(q, headerBlock);
    LOG_EXIT;
}

/////////////////////////////////////////////////////////////////////

static void DoFlushRequest(queue_t* q, mblk_t* mp)
    // Handle an M_FLUSH message as per "STREAMS Modules and Drivers", p8-12.
    //
    // Environment: write put routine
{
    TRACE_SETUP;

    LOG_ENTRY( "TPIFile:DoFlushRequest" );
    OTAssert("DoFlushRequest: Not the write queue", IsWriteQ(q) );

    // If we've been asked to flush the write queue, just go and do it.

    if ( ( *(mp->b_rptr) & FLUSHW ) != 0 ) {
        flushq(q, FLUSHDATA);
    }

    // Check whether we've been asked to flush the read queue.

    if ( ( *(mp->b_rptr) & FLUSHR ) != 0 ) {

        // If so, clear the FLUSHW bit and send the flush
        // request back up stream.  We don't actually flush
        // the contents of the read queue because doing so
        // might compromise the delicate state machine in
        // this module.  I need to investigate whether this
        // is a serious problem and, if it is, work out a fix.

        *(mp->b_rptr) &= ~FLUSHW;
        qreply(q, mp);

    } else {

        // We weren't asked to flush the read queue, so
        // we're done with this message.

        freemsg(mp);
    }
    LOG_EXIT;
}

/////////////////////////////////////////////////////////////////////
// Write-side put routine

static SInt32 TPIFileWritePut(queue_t* q, mblk_t* mp)
    // This routine is called by STREAMS when it has a message for our
    // module.  This routine is basically a big case statement that
    // dispatches to our various message handling routines.
    //
    // Note that a production module would probably want to handle
    // the high frequency requests (like T_DATA_REQ) inline in this
    // routine for maximum speed, but this module is still in the
    // "make it work" stage.
    //
    // Environment: standard STREAMS entry point
{
    TRACE_SETUP;
    PerStreamDataPtr streamData;
    struct iocblk *iocblkPtr;

    LOG_ENTRY( "TPIFile:TPIFileWritePut" );

    OTAssert("TPIFileWritePut: Not the write queue", IsWriteQ(q) );

    streamData = GetPerStreamData(q);

    switch ( mp->b_datap->db_type ) {
        case M_IOCTL:
            // "STREAMS Modules and Drivers" p8-33 says: "A driver must process
            // an M_IOCTL message.  Otherwise, the Stream head blocks for an M_IOCNAK
            // or M_IOCACK until the timeout (potentially infinite) expires. If a driver
            // does not understand an ioctl, an M_IOCNAK message must be sent to upstream."

            // I'm not sure why we're allowed to reuse this datab without
            // check it for read-only, but everyone does so!

            mp->b_datap->db_type = M_IOCNAK;
            iocblkPtr = (struct iocblk *) mp->b_rptr;
            iocblkPtr->ioc_error = EINVAL;

            qreply(q, mp);
            break;

        case M_DATA:
            // "STREAMS Modules and Drivers", Appendix A-2, T_DATA_REQ (7tpi) says:
            // "The transport provider must also recognize a message of one or more
            // M_DATA message blocks without the leading M_PROTO message block as a
            // T_DATA_REQ primitive. This message type will be initiated from the write
            // (BA_OS) operating system service routine."

            DoDataRequest(q, nil, mp);
            break;

        case M_FLUSH:
            DoFlushRequest(q, mp);
            break;

        default:
            switch ( GetPrimitive(mp) ) {
                case T_INFO_REQ:
                    DoInfoRequest(q, mp);
                    break;
                case T_BIND_REQ:
                    DoBindRequest(q, mp);
                    break;
                case T_CONN_REQ:
                    DoConnectRequest(q, mp);
                    break;
                case T_DISCON_REQ:
                    DoDisconnectRequest(q, mp);
                    break;
                case T_UNBIND_REQ:
                    DoUnbindRequest(q, mp);
                    break;
                case T_OPTMGMT_REQ:
                    DoOptionManagementRequest(q, mp);
                    break;
                case T_DATA_REQ:
                    DoDataRequest(q, mp, mp->b_cont);
                    break;

                // Standard TPI messages that are inappropriate.
                case T_ORDREL_REQ:
                case T_UNITDATA_REQ:
                case T_EXDATA_REQ:

                // Extended TPI messages that are inappropriate.
                case T_ADDR_REQ:

                // Transaction TPI messages that are inappropriate.
                case T_UREQUEST_REQ:
                case T_REQUEST_REQ:
                case T_UREPLY_REQ:
                case T_REPLY_REQ:
                case T_CANCELREQUEST_REQ:
                case T_CANCELREPLY_REQ:

                // Mapper TPI messages that are inappropriate.
                case T_REGNAME_REQ:
                case T_DELNAME_REQ:
                case T_LKUPNAME_REQ:

                    DoStreamFatalError(q, mp);
                    break;

                case kNoPrimitive:
                default:
                    // "STREAMS Modules and Drivers" p8-33 says: "Messages that are
                    // not understood by the driver should be freed."

                    OTDebugBreak("TPIFileWritePut: Message not understood, freeing");

                    freemsg(mp);
                    break;
            }
            break;
    }

    LOG_EXIT;

    return 0;
}

/////////////////////////////////////////////////////////////////////
// Read-side put routine

static SInt32 TPIFileReadPut(queue_t* q, mblk_t* mp)
    // Because we're a driver (ie at the end of the stream) this routine
    // should never be called by STREAMS.
    //
    // Environment: standard STREAMS entry point
{
    PerStreamDataPtr streamData;

    OTAssert("TPIFileReadPut: Not the read queue", IsReadQ(q) );

    streamData = GetPerStreamData(q);

    switch ( GetPrimitive(mp) ) {
        default:
            OTDebugBreak("TPIFileReadPut: Was called!");
            break;
    }

    return 0;
}

/////////////////////////////////////////////////////////////////////

static void StartDataIndication(queue_t* q)
    // StartDataIndication is called by either DoConnectionConfirm
    // or DoDataIndication to start a read request for a file.
    // The routine allocates a buffer for the T_DATA_IND message
    // and then starts an asynchronous File Manager _Read
    // request to read data into that buffer.  When the _Read
    // completes, CustomIOCompletion is called.  It puts
    // the T_DATA_IND message on the read-side queue, which schedules
    // TPIFileReadService, which in turn calls DoDataIndication to
    // send the data upstream, and then calls StartDataIndication
    // again to start the next block.
    //
    // Environment: read service routine
{
    TRACE_SETUP;
    PerStreamDataPtr streamData;
    mblk_t* mp = nil;
    mblk_t* dataBuffer = nil;

    LOG_ENTRY( "TPIFile:StartDataIndication" );

    OTAssert("StartDataIndication: Not the read queue", IsReadQ(q) );

    streamData = GetPerStreamData(q);
    OTAssert("StartDataIndication: Already have a current message", streamData->currentMessage == nil);
    OTAssert("StartDataIndication: Wrong state", streamData->currentState == TS_DATA_XFER);

    // Create a data buffer for the data to be read.

    dataBuffer = allocb(kDataBufferSize, 0);
    OTAssert("StartDataIndication: allocb failed", dataBuffer != nil);

    // Create a T_DATA_IND message with that data buffer.

    mp = qmi_tpi_data_ind(dataBuffer, 0, 0);
    OTAssert("StartDataIndication: qmi_tpi_data_ind failed", mp != nil);

    // Remember the T_DATA_IND message as the current message, ie the
    // one that CustomIOCompletion will operate on.

    streamData->currentMessage = mp;

    // Start the File Manager read request.

    streamData->fileParamBlock.ioParam.ioRefNum = streamData->fileRefNum;
    streamData->fileParamBlock.ioParam.ioBuffer = (char *) dataBuffer->b_datap->db_base;
    streamData->fileParamBlock.ioParam.ioReqCount = dataBuffer->b_datap->db_lim - dataBuffer->b_datap->db_base;
    streamData->fileParamBlock.ioParam.ioPosMode = fsAtMark + noCacheMask;
    streamData->fileParamBlock.ioParam.ioPosOffset = 0;
    streamData->fileParamBlock.ioParam.ioCompletion = gCustomIOCompletionUPP;
    (void) PBReadAsync( (ParmBlkPtr) &streamData->fileParamBlock);
    // Throw away error result from PBReadAsync because File Manager will still call
    //  our completion routine.

    // ... continue in CustomIOCompletion... TPIFileReadService... DoDataIndication...
    LOG_EXIT;
}

/////////////////////////////////////////////////////////////////////

static void DoConnectConfirm(queue_t* q, mblk_t* mp)
    // Handle processing of a T_CONN_CON message.  This message was originally
    // created in DoConnectRequest and put into the per-stream data
    // variable currentMessage.  When the ioCompletion for the PBHOpenAsync
    // fired, the CustomIOCompletion fired and queued the message on to the
    // read-side queue.  The TPIFileReadService then called this routine.
    //
    // Because the stream is still in TS_WCON_CREQ, the routine just sends
    // the T_CONN_CON upstream to indicate to the client that we have
    // a connection in place (or a T_DISCON_IND if the file failed to open).
    //
    // Environment: read service routine
{
    TRACE_SETUP;
    PerStreamDataPtr streamData;

    LOG_ENTRY( "TPIFile:DoConnectConfirm" );

    OTAssert("DoConnectConfirm: Not the read queue", IsReadQ(q) );

    streamData = GetPerStreamData(q);
    OTAssert("DoConnectConfirm: Already have a current message", streamData->currentMessage == nil);

    OTAssert("DoConnectConfirm: Wrong state", streamData->currentState == TS_WCON_CREQ);

    // This is the final stage of the connection process.  If the
    // PBHOpenAsync failed, free the T_CONN_CON and send a T_DISCON_IND
    // upstream instead.
    // If the open succeed, start the process of data transfer and send
    // the T_CONN_CON upstream.

    if (streamData->fileParamBlock.fileParam.ioResult != noErr) {
        freemsg(mp);
        mp = mi_tpi_discon_ind(nil, streamData->fileParamBlock.fileParam.ioResult, -1);
        OTAssert("DoConnectConfirm: mi_tpi_discon_ind failed", mp != nil);
        streamData->currentState = TS_IDLE;
    } else {
        streamData->fileRefNum = streamData->fileParamBlock.fileParam.ioFRefNum;
        streamData->currentState = TS_DATA_XFER;
        StartDataIndication(q);
    }
    putnext(q, mp);
    LOG_EXIT;
}

/////////////////////////////////////////////////////////////////////

static void DoDataIndication(queue_t* q, mblk_t* mp)
    // Handle processing of a T_DATA_IND message.  This message was originally
    // created in StartDataIndication and put into the per-stream data
    // variable currentMessage.  When the ioCompletion for the PBReadAsync
    // fired, the CustomIOCompletion fired and queued the message on to the
    // read-side queue.  The TPIFileReadService then called this routine.
    //
    // Because the stream is still in TS_DATA_XFER, the routine just sends
    // the T_DATA_IND upstream to indicate to the client the arrival of
    // new data (or a T_DISCON_IND if the read failed).
    //
    // Environment: read service routine
{
    TRACE_SETUP;
    PerStreamDataPtr streamData;

    LOG_ENTRY( "TPIFile:DoDataIndication" );

    OTAssert("DoDataIndication: Not the read queue", IsReadQ(q) );

    streamData = GetPerStreamData(q);
    OTAssert("DoDataIndication: Already have a current message", streamData->currentMessage == nil);

    OTAssert("DoDataIndication: Wrong state", streamData->currentState == TS_DATA_XFER);

    // If we get an eofErr then check whether we read any data at all
    // -- if we did, return the data to the user and then start another
    // read attempt, which will fail and finally cause the T_DISCON_IND to be
    // issued

    if (streamData->fileParamBlock.ioParam.ioResult == eofErr &&
            streamData->fileParamBlock.ioParam.ioActCount != 0) {
        streamData->fileParamBlock.ioParam.ioResult = noErr;
    }

    // If we get an error, free the T_DATA_IND and send a T_DISCON_IND upstream instead.

    if (streamData->fileParamBlock.ioParam.ioResult != noErr) {
        freemsg(mp);
        mp = mi_tpi_discon_ind(nil, streamData->fileParamBlock.ioParam.ioResult, -1);
        OTAssert("DoDataIndication: mi_tpi_discon_ind failed", mp != nil);

        CloseFileAndPostMessage(q, mp);

        // ... continue in CustomIOCompletion... TPIFileReadService... DoDisconnectIndication...

    } else {
        OTAssert("DoDataIndication: No data in data message", mp->b_cont != nil);
        mp->b_cont->b_rptr = mp->b_cont->b_datap->db_base;
        mp->b_cont->b_wptr = mp->b_cont->b_datap->db_base + streamData->fileParamBlock.ioParam.ioActCount;
        putnext(q, mp);
        StartDataIndication(q);
    }

    LOG_EXIT;
}

/////////////////////////////////////////////////////////////////////

static void DoDisconnectRequestAck(queue_t* q, mblk_t* mp)
    // This routine is called when a T_OK_ACK message shows up on
    // the read-side queue.  This message was put there
    // by DoDisconnectRequest, which also switched the state to either
    // TS_WACK_DREQ9 or TS_WACK_DREQ6 depending on the state that
    // the stream was in when the T_DISCON_REQ arrived.
    //
    // See the comments in each branch of the case statement for
    // how these different cases are handled.
    //
    // Environment: read service routine
{
    TRACE_SETUP;
    PerStreamDataPtr streamData;

    LOG_ENTRY( "TPIFile:DoDisconnectRequestAck" );

    OTAssert("DoDisconnectRequestAck: Not the read queue", IsReadQ(q) );

    OTAssert("DoDisconnectRequestAck: Unexpected message primitive",
            ( (GetPrimitive(mp) == T_OK_ACK) && (( (struct T_ok_ack *) mp->b_rptr)->CORRECT_prim == T_DISCON_REQ) )
        );

    streamData = GetPerStreamData(q);
    OTAssert("DoDisconnectRequestAck: Already have a current message", streamData->currentMessage == nil);

    switch ( streamData->currentState ) {
        case TS_WACK_DREQ6:
            // We've been asked to disconnect during the connection process,
            // ie between when we got the T_CONN_REQ and when the PBHOpenAsync
            // completed.  In this case, we ACK the T_DISCON_REQ and go back to
            // TS_IDLE state.  Except it's a bit more complicated than that...

            switch ( streamData->wackSubstate ) {

                case WSS_OPEN:
                    // This is the first time we've hit this routine.  Check to see
                    // whether the file opened successfully.  If it did, we have
                    // to organise to close it before sending up the T_OK_ACK.
                    // We switch the wackSubState to WSS_ISSUE_MP to ensure that,
                    // the next time we come to this routine, we know that we've
                    // already closed the file.

                    OTAssert("DoDisconnectRequestAck: File should be closed", streamData->fileRefNum == 0);
                    if (streamData->fileParamBlock.fileParam.ioResult != noErr) {
                        // If we got an error from the PBHOpenAsync, the file is not open, so we can
                        // just ACK the disconnect request immediately.

                        streamData->currentState = TS_IDLE;
                        putnext(q, mp);

                    } else {

                        // The file is now open, but we have to close it before we
                        // can ACK the disconnect request.  So start the async
                        // close request.

                        streamData->fileRefNum = streamData->fileParamBlock.fileParam.ioFRefNum;

                        streamData->wackSubstate = WSS_ISSUE_MP;
                        CloseFileAndPostMessage(q, mp);
                        // ... continue in CustomIOCompletion... TPIFileReadService... DoDisconnectRequestAck
                        // ie the WSS_ISSUE_MP branch of this case statement
                    }
                    break;

                case WSS_ISSUE_MP:
                    // This is the second time we've been called.  We've closed the
                    // file, so just go ahead and send the T_OK_ACK upstream to *finally*
                    // ACK the T_DISCON_REQ.

                    OTAssert("DoDisconnectRequestAck: PBClose failed with error", streamData->fileParamBlock.fileParam.ioResult == noErr);
                    OTAssert("DoDisconnectRequestAck: File should be closed", streamData->fileRefNum == 0);
                    streamData->currentState = TS_IDLE;
                    putnext(q, mp);
                    break;

                default:
                    OTDebugBreak("DoDisconnectRequestAck: wackSubstate out of range");
                    break;
            }
            break;

        case TS_WACK_DREQ9:
            // If the stream is in TS_WACK_DREQ9, then we've received a T_DISCON_REQ
            // while we were in the TS_DATA_XFER state, ie we've been asked to disconnect
            // during the reading process.  Any pending reads have now
            // completed, so we can just send the T_OK_ACK upstream to ACK the
            // T_DISCON_REQ.  Well almost )-:  If the file is open we must close
            // it before doing this.

            if ( streamData->fileRefNum != 0 ) {
                CloseFileAndPostMessage(q, mp);
                // ... continue in CustomIOCompletion... TPIFileReadService... DoDisconnectRequestAck
                // ie the other branch of this if statement
            } else {
                streamData->currentState = TS_IDLE;
                putnext(q, mp);
            }

            break;

        default:
            OTDebugBreak("DoDisconnectRequestAck: Unexpected state");
            break;
    }
    LOG_EXIT;
}

/////////////////////////////////////////////////////////////////////

static void DoDisconnectIndication(queue_t* q, mblk_t* mp)
    // This routine is called when a T_DISCON_IND message shows up
    // on the read-side queue.  This message was put there by
    // DoDataIndication in response to running into an error reading
    // the file.  We just switch states and forward the message
    // upstream.
    //
    // Environment: read service routine
{
    TRACE_SETUP;
    PerStreamDataPtr streamData;

    LOG_ENTRY( "TPIFile:DoDisconnectIndication" );

    OTAssert("DoDisconnectIndication: Not the read queue", IsReadQ(q) );

    OTAssert("DoDisconnectIndication: Unexpected message primitive",
            ( (GetPrimitive(mp) == T_DISCON_IND) )
        );

    streamData = GetPerStreamData(q);
    OTAssert("DoDisconnectIndication: currentMessage should be nil", streamData->currentMessage == nil);

    OTAssert("DoDisconnectIndication: Wrong state", streamData->currentState == TS_DATA_XFER);
    OTAssert("DoDisconnectIndication: File should be closed", streamData->fileRefNum == 0);

    streamData->currentState = TS_IDLE;
    putnext(q, mp);
    LOG_EXIT;
}

/////////////////////////////////////////////////////////////////////
// Read-side service routine

static SInt32 TPIFileReadService(queue_t* q)
    // This routine is called by STREAMS when someone puts a message
    // on the read-side queue.  Seeing as we're a driver, the only person
    // who should be putting messages on our queue.  Some of these messages
    // have a special meaning, more than what the TPI defines, so we can't
    // just unilaterally send them upstream.  We have to dispatch them out
    // to the appropriate handler, which deals with the special cases.
    //
    // For example, when the completion routine for a PBReadAsync fires,
    // it simply queues the T_DATA_IND message on this queue.  However
    // no one has checked whether the read was successful yet, so we
    // can't just forward the T_DATA_IND upstream.  Instead this routine
    // sees the T_DATA_IND and forwards it to DoDataIndication, which
    // checks the results of the read and does the right thing.
    //
    // Also note that, by handling the data transfer operations through
    // a this queue, we automatically become a STREAMS flow control
    // good citizen, just by following the standard STREAMS structure
    // for a service routine.
    //
    // Environment: read service routine
{
    TRACE_SETUP;
    mblk_t *mp;

    LOG_ENTRY( "TPIFile:TPIFileReadService" );

    OTAssert("TPIFileReadService: Not the read queue", IsReadQ(q) );

    // Standard STREAMS flow control structure.  Don't blame me for its bad style (-:

    while ( (mp = getq(q)) != nil ) {

        if ((mp->b_datap->db_type < QPCTL) && !canputnext(q)) {
            putbq(q, mp);
            goto done;
        }

        switch ( GetPrimitive(mp) ) {
            case T_CONN_CON:
                DoConnectConfirm(q, mp);
                break;
            case T_DATA_IND:
                DoDataIndication(q, mp);
                break;
            case T_OK_ACK:
                DoDisconnectRequestAck(q, mp);
                break;
            case T_DISCON_IND:
                DoDisconnectIndication(q, mp);
                break;
            default:
                OTDebugBreak("TPIFileReadService: Unexpected type");
                break;
        }

    }

done:
    LOG_EXIT;
    return (0);
}

/////////////////////////////////////////////////////////////////////
// Static Declaration Structures

static struct module_info gModuleInfo =
{
    9990,                // Module Number, only useful for debugging
    "TPIFile",            // Name of module
    0,                    // Minimum data size
    INFPSZ,                // Maximum data size
    65536,                // Hi water mark for queue
    32768                // Lo water mark for queue
};

static struct qinit gReadInit =
{
    TPIFileReadPut,        // Put routine for "incoming" data
    TPIFileReadService,    // Service routine for "incoming" data
    TPIFileOpen,        // Our open routine
    TPIFileClose,         // Our close routine
    nil,                // No admin routine
    &gModuleInfo        // Our module_info
};

static struct qinit gWriteInit =
{
    TPIFileWritePut,    // Put routine for client data
    nil,                // Service routine for client data
    nil,                // open  field only used in read-side structure
    nil,                // close field only used in read-side structure
    nil,                // admin field only used in read-side structure
    &gModuleInfo        // Our module_info
};

static struct streamtab theStreamTab =
{
    &gReadInit,            // Our read-side qinit structure
    &gWriteInit,        // Our write-side qinit structure
    0,                    // We are not a mux, so set this to nil
    0                    // We are not a mux, so set this to nil
};

/////////////////////////////////////////////////////////////////////
// Macintosh-specific Static Structures

static struct install_info theInstallInfo =
{
    &theStreamTab,            // Stream Tab pointer
    kOTModIsDriver + kOTModUpperIsTPI,
                            // Tell OT that we are a driver, not a module
    SQLVL_MODULE,            // Synchronization level, module level for the moment
    0,                        // Shared writer list buddy
    0,                        // Open Transport use - always set to 0
    0                        // Flag - always set to 0
};

// Prototypes for the exported routines below.

extern Boolean InitStreamModule(TPIFilePortInfoRecordPtr portInfo);
extern void TerminateStreamModule(void);
extern install_info* GetOTInstallInfo();

#pragma export list InitStreamModule, TerminateStreamModule, GetOTInstallInfo

// Export entry point

extern Boolean InitStreamModule(TPIFilePortInfoRecordPtr portInfo)
    // Initialises the module.  Always called at SystemTask time,
    // so we can call NewIOCompletionProc.
{
    TRACE_SETUP;
    Boolean result;

    OTDebugBreak("TPIFile: InitStreamModule");

    LOG_ENTRY( "TPIFile:InitStreamModule" );
    OTAssert("InitStreamModule: Bad magic in TPIFilePortInfoRecord",
                        (portInfo->magic1 == kTPIFilePortInfoMagic1) &&
                        (portInfo->magic2 == kTPIFilePortInfoMagic2)
            );
    #if ! qDebug
        // portInfo is only used in the above assertion, which is compiled out
        // if we're not debugging.
        #pragma unused(portInfo)
    #endif

    gCustomIOCompletionUPP = NewIOCompletionProc(CustomIOCompletion);
    gCleanUpAnyDetachedStreamsTaskID = OTCreateDeferredTask(CleanUpAnyDetachedStreams, nil);

    result = (gCustomIOCompletionUPP != nil && gCleanUpAnyDetachedStreamsTaskID != 0);

    LOG_EXIT;
    return (result);
}

extern void TerminateStreamModule(void)
    // Shuts down the module.  Always called at SystemTask time,
    // so we can call DisposeRoutineDescriptor.
{
    TRACE_SETUP;
    OSStatus err;

    LOG_ENTRY( "TPIFile:TerminateStreamModule" );

    if (gCustomIOCompletionUPP != nil) {
        DisposeRoutineDescriptor(gCustomIOCompletionUPP);
        gCustomIOCompletionUPP = nil;
    }
    if (gCleanUpAnyDetachedStreamsTaskID != 0) {
        err = OTDestroyDeferredTask(gCleanUpAnyDetachedStreamsTaskID);
        OTAssert("TerminateStreamModule: OTDestroyDeferredTask failed", err == noErr);
        gCleanUpAnyDetachedStreamsTaskID = 0;
    }

    // The purpose of gCleanUpAnyDetachedStreamsTaskID is to clean up any
    // detached streams that are pending the completetion of an asynchronous
    // _Close.  Unfortunately, in the case where the last stream is closed,
    // the TerminateStreamModule routine will be executed before the deferred
    // task runs, so we have just destroyed our last chance of cleaning up
    // the stream.  We obviously have to avoid this case, so we sit here
    // waiting for those _Close's to complete.  This is legal because we are
    // allowed to block inside the TerminateStreamModule routine.
    //
    // Well almost...  If the file we're accessing is on AppleShare, this
    // causes the machine to deadlock.  This is because we're waiting for
    // PBCloseAsync to complete, but it can't complete until we leave this
    // routine because AppleShare relies on OT to deliver packets to
    // complete the request and OT won't be delivering packets until we
    // leave this routine.  The upshot is that we deadlock.

    while ( gModuleList != nil ) {
        CleanUpAnyDetachedStreams(nil);
    }

    // While this assert can never trigger on my module (because of the previous
    // while loop), it's an excellent idea to have one of these in your code.

    OTAssert("TerminateStreamModule: Streams are still active", gModuleList == nil);

    LOG_EXIT;
}

extern install_info* GetOTInstallInfo()
    // Return pointer to install_info to STREAMS.
{
    return &theInstallInfo;
}